package demo.util.http;

import com.alibaba.fastjson.JSONObject;
import demo.designPattern.observer.RxJavaDemo;
import demo.java.lang.ThreadDemo;
import demo.vo.common.TimeCost;
import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.BackpressureStrategy;
import io.reactivex.rxjava3.core.Flowable;
import io.reactivex.rxjava3.core.FlowableEmitter;
import io.reactivex.rxjava3.core.FlowableOnSubscribe;
import okhttp3.*;
import okhttp3.sse.EventSource;
import okhttp3.sse.EventSourceListener;
import okhttp3.sse.EventSources;
import okio.BufferedSink;
import okio.BufferedSource;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 主要特点
 * <li>同步和异步请求：
 * 同步请求会在当前线程等待响应，适合不需要并发的简单请求。
 * 异步请求会将网络操作交由后台线程处理，不会阻塞主线程，适合需要并发处理或在 Android 等环境中使用。
 *
 * <li>连接池： OkHttp 默认会使用连接池来复用 HTTP 连接，从而提高性能，减少连接的建立和关闭的开销。
 * <li>拦截器 (Interceptor)： 拦截器允许在请求和响应时进行操作，例如可以在请求发送前添加认证信息，或在响应到达后进行日志记录。
 * <li>自动处理 HTTP/2 和 SPDY： OkHttp 默认支持 HTTP/2 协议，可以提升多路复用性能，使多个请求共享一个 TCP 连接。
 * <li>缓存机制： OkHttp 提供了默认的缓存机制，可以根据 HTTP 响应头自动缓存请求结果，减少重复网络请求。
 * <li>超时控制： 可以对连接、读取和写入操作分别设置超时，避免长时间无响应的请求卡住应用。
 */
public class OkHttpClientDemo {

    public static final String octet_stream_str = "application/octet-stream";
    public static final MediaType octet_stream = MediaType.parse(octet_stream_str);
    static final Logger logger = LoggerFactory.getLogger(OkHttpClientDemo.class);
    private static final String MIME_JSON = "application/json; charset=utf-8";
    /**
     * 日志拦截器，打印请求和响应信息
     */
    static Interceptor loggerInterceptor = new Interceptor() {
        @NotNull
        @Override
        public Response intercept(@NotNull Chain chain) throws IOException {
            TimeCost t0 = new TimeCost();
            Request req = chain.request();
            String method = req.method();
            HttpUrl url = req.url();
            String host = url.host();
            int code = 0;
            Response res;
            try {
                res = chain.proceed(chain.request());
                code = res.code();
            } catch (IOException e) {
                logger.error("error", e);
                throw e;
            } finally {
                logger.info("{} {} return {},cost {} ms", method, host, code, t0.cost());
                logger.info("request {} return {},cost {} ms",  url, code, t0.cost());
            }

            return res;
        }
    };

    static final OkHttpClient OK_HTTP_CLIENT = new OkHttpClient.Builder()
            .connectTimeout(3, TimeUnit.SECONDS)
            .writeTimeout(50, TimeUnit.SECONDS)
            .readTimeout(50, TimeUnit.SECONDS)
            .addInterceptor(loggerInterceptor)
            .build();

    // POST-json传参
    public static String postJsonWithHeader(String url, Object reqBody, Map<String, String> headers) throws IOException {
        Request.Builder request = new Request.Builder()
                .url(url)
                .header("Accept", MIME_JSON)
                .post(FormBody.create(MediaType.parse(MIME_JSON), JSONObject.toJSONString(reqBody)));
        // 遍历并添加 header
        if (headers != null) {
            headers.forEach(request::header);
        }

        try (Response response = OK_HTTP_CLIENT.newCall(request.build()).execute()) {
            logger.info("response_code:{}, response_body:{}", response.code(), response.body().string());
            return response.body().string();
        }
    }

    // POST-param传参
    public static String postParamWithHeader(String url, Map<String, Object> paramMap, Map<String, String> headers) throws IOException {
        FormBody.Builder formBody = new FormBody.Builder();
        if (Objects.nonNull(paramMap)) {
            paramMap.forEach((k, v) -> formBody.add(k, (String) v));
        }

        Request.Builder request = new Request.Builder()
                .url(url)
                .header("Accept", MIME_JSON)
                .post(formBody.build());

        // 遍历并添加 header
        if (headers != null) {
            headers.forEach(request::header);
        }

        try (Response response = OK_HTTP_CLIENT.newCall(request.build()).execute()) {
            logger.info("response_code:{}, response_body:{}", response.code(), response.body().string());
            return response.body().string();
        }
    }

    public static void putStream(String url, InputStream inputStream, long contentLength) {
        Response response;
        try {
            RequestBody requestBody2 = null;
//                requestBody2    = new FormBody(inputStream, contentLength);

            Request request = new Request.Builder().url(url)
                    .addHeader("Content-Type", octet_stream_str)
                    .put(requestBody2)
                    .build();
            response = OK_HTTP_CLIENT.newCall(request).execute();

            String result = response.body().string();
            if (response.code() != 200) {
                throw new Exception("Saturn SDK stream upload failed, url:" + url + "response.code():" + response.code() + ", msg:" + result);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            closeQuietly(inputStream);
        }
    }

    /**
     * 上传文件
     *
     * @param url
     * @param file
     * @return
     * @throws Exception
     */
    public static String putFile(String url, File file) throws Exception {
        RequestBody requestBody = RequestBody.create(octet_stream, file);
        Request request = new Request.Builder().url(url)
                .addHeader("Content-Type", octet_stream_str)
                .put(requestBody).build();

        Response response = OK_HTTP_CLIENT.newCall(request).execute();
        String result = response.body().string();
        int code = response.code();
        logger.info("putFile end, url:{}, response.code:{}, result:{}", url, code, result);
        if (code != 200) {
            throw new Exception("putFile failed, response.code():" + code + ", result:" + result);
        }
        return result;
    }

    private static String basicDelete(String url, Headers headers) throws IOException {
        Request.Builder builder = new Request.Builder().url(url).delete();
        if (headers != null) {
            builder.headers(headers);
        }
        try (Response response = OK_HTTP_CLIENT.newCall(builder.build()).execute()) {
            return response.body().string();
        }
    }

    public static void closeQuietly(Closeable is) {
        if (is != null) {
            try {
                is.close();
            } catch (Exception ex) {
                logger.error("Resources encounter an exception when closing，ex：{}", ex.getMessage());
            }
        }
    }

    @Test
    public void testEnqueuePost() throws IOException {
        OkHttpClient client = new OkHttpClient().newBuilder()
                .build();
        MediaType mediaType = MediaType.parse("application/json");
        RequestBody requestBody = RequestBody.create(mediaType, "{}");
        Request request = new Request.Builder()
                .url("http://ops.ximalaya.com/ai-cloud-deploy-server/k8s/serving/group/instance/getPodInfo")
                .method("POST", requestBody)
                .addHeader("User-Agent", "Apifox/1.0.0 (https://apifox.com)")
                .addHeader("Content-Type", "application/json")
                .build();
        Call call = client.newCall(request);
        call.enqueue(new Callback() {
            @Override
            public void onResponse(@NotNull Call call, @NotNull Response response) throws IOException {

            }

            @Override
            public void onFailure(@NotNull Call call, @NotNull IOException e) {

            }
        });
        Response response = call.execute();
        System.out.println(response);
        ResponseBody responseBody = response.body();
        if (responseBody != null) {
            System.out.println(responseBody.string());
        }
    }

    /**
     * 同步请求
     *
     * @throws IOException
     */
    @Test
    public void testExecuteGet() throws IOException {
        String url = "http://xdcs.ximalaya.com/logClassifyRule/listAll";
        Request.Builder builder = new Request.Builder()
                .url(url)
                .get();
        String result;
        Call call = OK_HTTP_CLIENT.newCall(builder.build());
        try (Response response = call.execute()) {
            ResponseBody body = response.body();
            if (body == null) {
                result = null;
            } else {
                result = body.string();
            }
        }
        System.out.println(result);
    }

    /**
     * 异步请求
     *
     * @throws IOException
     */
    @Test
    public void testEnqueueGet() throws IOException {
        String url = "http://xdcs.ximalaya.com/logClassifyRule/listAll";
        Request.Builder builder = new Request.Builder()
                .url(url)
                .get();
        Call call = OK_HTTP_CLIENT.newCall(builder.build());

        call.enqueue(new Callback() {
            @Override
            public void onResponse(@NotNull Call call, @NotNull Response response) throws IOException {
                ResponseBody body = response.body();
                if (body != null) {
                    logger.info(body.string());
                }
            }

            @Override
            public void onFailure(@NotNull Call call, @NotNull IOException e) {
                logger.info("Failure:" + e.getMessage());
            }
        });

        ThreadDemo.safeSleep(2000);
    }

    /**
     * 流式请求
     */
    @Test
    public void testStreamRequest() {
        RequestBody requestBody = new RequestBody() {

            @Override
            public void writeTo(@NotNull BufferedSink bufferedSink) throws IOException {
                // 逐步写入数据
                bufferedSink.writeUtf8("First part of the data\n");
                bufferedSink.writeUtf8("Second part of the data\n");
                // 可以在这里进行其他写入操作
            }

            @Nullable
            @Override
            public MediaType contentType() {
                return MediaType.parse("text/plain; charset=utf-8");
            }
        };

        Request request = new Request.Builder()
                .url("https://www.example.com/upload")
                .post(requestBody) // 使用流式请求体
                .build();

        try (Response response = OK_HTTP_CLIENT.newCall(request).execute()) {
            if (response.isSuccessful()) {
                System.out.println("Request successful: " + response.body().string());
            } else {
                System.out.println("Request failed: " + response.message());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 流式响应
     */
    @Test
    public void testStreamResponse() {
        Request request = new Request.Builder()
                .url("http://172.26.9.10:12382/ai-infra-test/test/sse/time/9")
                .build();

        try (Response response = OK_HTTP_CLIENT.newCall(request).execute()) {
            if (response.isSuccessful()) {
                // 使用 BufferedSource 逐步读取响应体
                BufferedSource source = response.body().source();
                while (!source.exhausted()) {
                    String line = source.readUtf8Line();
                    System.out.println(line);
                }
            } else {
                System.out.println("Request failed: " + response.message());
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Test
    public void testSSE() {
        String url = "http://172.26.9.10:12382/ai-infra-test/test/sse/time/13";  // 替换为您的 SSE 接口 URL
        // 创建 OkHttpClient
        OkHttpClient client = new OkHttpClient();
        Request request = new Request.Builder()
                .url(url)
                .addHeader("Accept", "text/event-stream") // 设置接收类型为 SSE
                .build();

        // 发送请求
        client.newCall(request).enqueue(new Callback() {
            @Override
            public void onFailure(Call call, IOException e) {
                logger.error("onFailure Exception: {}", e);
            }

            @Override
            public void onResponse(Call call, Response response) throws IOException {
                if (!response.isSuccessful()) {
                    throw new IOException("Unexpected code " + response);
                }

                // 读取服务端推送的 SSE 数据
                try (BufferedReader reader = new BufferedReader(new InputStreamReader(response.body().byteStream()))) {
                    String line;
                    while ((line = reader.readLine()) != null) {
                        logger.info("Received: {}", line);
                    }
                }
            }
        });

        // 为了让程序持续运行，以便接收事件
        ThreadDemo.safeSleep(10_000);
    }

    /**
     * 这个方法在处理类似于 Server-Sent Events（SSE）这种长连接、实时数据推送场景时非常有用。
     * <p>
     * 什么是 Server-Sent Events (SSE)?
     * Server-Sent Events 是一种服务器推送技术，允许服务器通过持久连接向客户端发送实时更新。
     * 与 WebSocket 相比，SSE 是单向的，服务器可以向客户端发送数据，但客户端不能直接向服务器发送数据（虽然可以通过其他 HTTP 请求）。
     * <p>
     * EventSource.Factory 是一个工厂类，用于创建 EventSource 实例。
     * EventSource 代表一个连接至服务端推送的事件源。当服务器发送事件时，会触发对应的回调。
     * <p>
     * EventSources.createFactory() 方法用于创建一个 EventSource.Factory 实例，通常需要一个 OkHttpClient 和一个 EventSourceListener。
     * 它可以用于处理来自服务器的事件和响应。
     * <p>
     * 关键点
     * <li>创建连接：使用 EventSources.createFactory(client) 创建 EventSource.Factory，并通过 newEventSource 方法创建连接到服务器的 EventSource 实例。
     * <li>事件处理：通过实现 EventSourceListener，可以处理来自服务器的事件、连接失败和断开等情况。
     * <li>实时数据：适用于需要实时或近实时数据推送的场景，例如消息通知、状态更新等。
     */
    @Test
    public void testEventSource() {

        String url = "http://172.26.9.10:12382/ai-infra-test/test/sse/time/5";

        FlowableOnSubscribe<String> source = new FlowableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull FlowableEmitter<String> emitter) throws Throwable {
                EventSource.Factory factory = EventSources.createFactory(OK_HTTP_CLIENT);
                TimeCost t0 = new TimeCost();
                EventSourceListener eventSourceListener = new EventSourceListener() {

                    @Override
                    public void onOpen(@NotNull EventSource eventSource, @NotNull Response response) {
                        logger.info("onOpen response: {}", response);
                        super.onOpen(eventSource, response);
                    }

                    @Override
                    public void onEvent(EventSource eventSource, String id, String event, String data) {
                        // 处理收到的数据
                        logger.info("onEvent id: {}, event: {}, data: {}, cost:{} ms", id, event, data, t0.cost());
                        emitter.onNext(data);
                    }

                    @Override
                    public void onFailure(EventSource eventSource, Throwable t, Response response) {
                        // 处理连接失败
                        logger.error("onFailure Exception: {}, response:{}, cost {} ms", t, response, t0.cost());
                        System.err.println("Error: " + t.getMessage());
                        emitter.onError(t);
                    }

                    public void onClosed(EventSource eventSource) {
                        // 处理连接关闭
                        logger.info("onClosed eventSource: {}, cost:{} ms", eventSource, t0.cost());
                        emitter.onComplete();
                    }

                };

                Request request = new Request.Builder().url(url).build();
                EventSource eventSource = factory.newEventSource(request, eventSourceListener);
            }
        };

        Flowable<String> flowable = Flowable.create(source, BackpressureStrategy.BUFFER);
        // 订阅并处理流中的数据
        flowable.subscribe(RxJavaDemo.subscriber);

        ThreadDemo.safeSleep(10_000);
    }


}
